package io.dy.mq;
import io.dy.utils.ThreadUtils;

import java.util.LinkedList;

/**
 * @author dmj
 * @date 2024年03月25日 下午1:48
 * 消息队列
 */
public class MessageQueue<K,T> {

    private final LinkedList<Message<K,T>> queue = new LinkedList<>();

    private final Integer capacity;

    public MessageQueue(Integer capacity) {
        this.capacity = capacity;
    }

    /**
     * 获取消息
     * @return
     */
    public synchronized Message<K,T> take(){
        while (queue.isEmpty()){
            try {
                ThreadUtils.printThreadLog("队列为空，消费线程等待");
                this.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        Message<K, T> message = queue.removeFirst();
        ThreadUtils.printThreadLog("消费者获取消息 :" + message.toString());
        this.notifyAll();
        return message;
    }


    /**
     * 存入消息
     * @param msg
     */
    public synchronized void put(Message<K,T> msg){
        //判断队列容量是否已满 已满则等待
        while (queue.size() >= capacity){
            try {
                ThreadUtils.printThreadLog("队列已满，生产线程等待");
                this.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        ThreadUtils.printThreadLog("生产者生产数据 :" + msg.toString());
        queue.addLast(msg);
        this.notifyAll();
    }
}
